package com.lxl.testHd.etl.three;

import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IOUtils;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import java.io.BufferedReader;
import java.io.FileInputStream;
import java.io.IOException;
import java.io.InputStreamReader;
import java.net.URI;
import java.util.*;

/**
 * @author ：e_lixilin
 * @date ：2022/2/23 8:00
 * @description：
 * @modified By：
 */
public class CarMrCache {
    static class TollStationMapper extends Mapper<LongWritable, Text, Text, CarBean>{

        Text outKey = new Text();
        CarBean bean = new CarBean();
        @Override
        protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
            // 车辆行驶数据: 2018-09-15.05:56:48	2018-09-15.08:53:14	208	4	云AUJ791
            String[] fields = value.toString().split("\t");
            if (fields.length == 5) {
                bean.setBean(fields[0],fields[1],Long.parseLong(fields[2]),fields[4],"",0.0f,0);
                outKey.set(fields[3]);
                context.write(outKey, bean);
            }
        }
    }

    static class TollStationReducer extends Reducer<Text, CarBean, Text, NullWritable>{
        Map<String, CarBean> cacheMap = new HashMap<String, CarBean>();
        @Override
        protected void setup(Context context) throws IOException, InterruptedException {
            // 缓存小表数据
            URI cacheFile = context.getCacheFiles()[0];
            String path = cacheFile.getPath();
            BufferedReader reader = new BufferedReader(new InputStreamReader(new FileInputStream(path)));
            String line;
            while (StringUtils.isNotEmpty(line = reader.readLine())) {
                // 1	小型	1.5
                String[] fields = line.split("\t");
                CarBean car = new CarBean();
                car.setPl(Float.parseFloat(fields[2]));
                car.setTypeName(fields[1]);
                cacheMap.put(fields[0], car);
            }
            // 关闭流
            IOUtils.closeStream(reader);
        }
        Text outKey = new Text();
        StringBuilder sb = new StringBuilder();
        NullWritable outVal = NullWritable.get();
        @Override
        protected void reduce(Text key, Iterable<CarBean> values, Context context) throws IOException, InterruptedException {

            CarBean typeBean = cacheMap.get(key.toString());
            if (Objects.isNull(typeBean)) {
                typeBean = new CarBean();
                typeBean.setTypeName("xxx");
            }
            for (CarBean carBean : values) {
                sb.setLength(0);
                sb.append(typeBean.getTypeName());
                sb.append("\t");
                sb.append(carBean.getDistance());
                sb.append("\t");
                sb.append(carBean.getCarPlate());
                outKey.set(sb.toString());
                // 写出数据
                context.write(outKey,outVal);
            }
        }
    }


    public static void main(String[] args) throws Exception {
        args = new String[]{"D:\\big-data\\etl\\join", "D:\\big-data\\etl\\join\\outputcache"};
        Configuration conf = new Configuration();
        Job job = Job.getInstance(conf);
        job.setJarByClass(CarMrCache.class);
        job.setMapperClass(TollStationMapper.class);
        job.setReducerClass(TollStationReducer.class);

        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(CarBean.class);

        job.addCacheFile(new URI("file:///D:/big-data/etl/cartype/carType.txt"));

        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(NullWritable.class);
        FileInputFormat.setInputPaths(job, new Path(args[0]));
        FileOutputFormat.setOutputPath(job, new Path(args[1]));
        boolean res = job.waitForCompletion(true);
        System.exit(res?0:1);
    }
}
